"""
Case Type   : python驱动psycopg2
Case Name   : 占位符与非占位符一致时，sql功能正常
Create At   : 2024/4/2
Owner       : @ningyali
Description :
    1.解压openGauss-*-CentOS-x86_64-Python.tar.gz得到lib,替换psycopg2
    2.配置白名单
    3.配置lib
    4.连接数据库
    5.执行sql，sql中占位符为%
    6.执行sql，sql中占位符为$
    7.执行sql，sql中占位符为$，含非占位符%
    8.执行sql，sql中占位符为%，含非占位符%
    9.执行sql，sql中占位符与转义符一致为%
Expect      :
    1.拷贝成功
    2.配置成功
    3.配置成功
    4.连接数据库成功
    5.执行成功
    6.执行成功
    7.部分合理报错
    8.部分合理报错
    9.部分合理报错
History     :
"""
import os
import unittest

from testcase.CONNECTORS.PDBC.PSYCOPG2. \
    Opengauss_Function_Connect_Psycopg_Common import Psycopg2Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger


class Psycopg57(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.conn_info = {'dbname': self.pri_sh.node.db_name,
                          'user': self.pri_sh.node.db_user,
                          'password': self.pri_sh.node.db_password,
                          'host': self.pri_sh.node.ssh_host,
                          'port': self.pri_sh.node.db_port}
        self.psycopg_com = Psycopg2Common(**self.conn_info)

        text = '----前置操作: 备份白名单 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

    def test_1(self):
        text = '----step1: 解压PDBC得到lib,替换psycopg2 expect: 执行成功----'
        res = self.psycopg_com.decompress_pkg()
        self.assertTrue(res, f'执行失败: {text}')
        res = self.psycopg_com.set_new_psycopg2()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step2: 配置白名单 expect: 配置成功----'
        self.log.info(text)
        res = self.psycopg_com.set_pghba(open_flag=True)
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step3: 配置lib expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.set_lib_path()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step4: 连接数据库 expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.conn_db()
        expect = f"user={self.pri_sh.node.db_user} " \
                 f"dbname={self.pri_sh.node.db_name} " \
                 f"host={self.pri_sh.node.ssh_host} " \
                 f"port={self.pri_sh.node.db_port}"
        res = [self.assertIn(i, res) for i in expect.split()]
        self.assertEqual(len(res), 4, f'执行失败: {text}')

        text = '----step5:执行sql，sql中占位符为%  expect: 成功----'
        self.log.info(text)
        sql = "sql = \\\"select '1%s%s'\\\""
        run_sql = f'cur=conn.cursor()\n' \
                  f'{sql}\n' \
                  f'cur.execute(sql)\n' \
                  f'print(cur.fetchone())\n' \
                  f'cur.close()\n'
        res = self.psycopg_com.run_sql(sql=run_sql)
        self.assertIn("('1%s%s',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace("cur.execute(sql)",
                                   "cur.execute(sql, ((2, 3)), "
                                   "place_holder='%')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('123',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        text = '----step6:执行sql，sql中占位符为$  expect: 成功----'
        self.log.info(text)
        run_sql1 = run_sql.replace('%s%s', '\\\\\\$1\\\\\\$2')
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('1$1$2',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '\\\\\\$1\\\\\\$2').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2, 3)), place_holder='$')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('123',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        text = '----step7:执行sql，sql中占位符为$，含非占位符%  expect: 部分合理报错----'
        self.log.info(text)
        run_sql1 = run_sql.replace('%s%s', '%s\\\\\\$1')
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('1%s$1',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%s\\\\\\$1').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2,)), place_holder='$')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('1%s2',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%s\\\\\\$1').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2,3)), place_holder='$')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = 'TypeError: not all arguments converted ' \
                 'during string formatting'
        self.assertIn(expect, res, f'执行失败: {text}')

        text = '----step8:执行sql，sql中占位符为%，含非占位符%  expect: 部分合理报错----'
        self.log.info(text)
        run_sql1 = run_sql.replace('%s%s', '%%s')
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('1%%s',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%%s').replace(
            "cur.execute(sql)", "cur.execute(sql, (()))")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = 'IndexError: tuple index out of range'
        self.assertIn(expect, res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%%s').replace(
            "cur.execute(sql)", "cur.execute(sql, 2)")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = "TypeError: 'int' object does not support indexing"
        self.assertIn(expect, res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%%s').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2,)), place_holder='%')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = 'TypeError: not all arguments converted during ' \
                 'string formatting'
        self.assertIn(expect, res, f'执行失败: {text}')

        text = '----step9:执行sql，sql中占位符与转义符一致为%  expect: 部分合理报错----'
        self.log.info(text)
        run_sql1 = run_sql.replace('%s%s', '%%s%s')
        res = self.psycopg_com.run_sql(sql=run_sql1)
        self.assertIn("('1%%s%s',)", res, f'执行失败: {text}')
        self.assertNotIn('errors', res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%%s%s').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2,)), place_holder='%')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = 'IndexError: tuple index out of range'
        self.assertIn(expect, res, f'执行失败: {text}')

        run_sql1 = run_sql.replace('%s%s', '%%s%s').replace(
            "cur.execute(sql)", "cur.execute(sql, ((2,3)), place_holder='%')")
        res = self.psycopg_com.run_sql(sql=run_sql1)
        expect = 'TypeError: not all arguments converted ' \
                 'during string formatting'
        self.assertIn(expect, res, f'执行失败: {text}')

    def tearDown(self):
        text = '----run teardown----'
        self.log.info(text)

        text1 = f'----后置操作：删除解压文件目录----'
        self.log.info(text1)
        res1 = self.psycopg_com.clean_decompress_file()

        text2 = f'----后置操作：还原白名单----'
        self.log.info(text2)
        res2 = self.psycopg_com.bak_conf(bak_flag=False)

        self.assertTrue(res1, f'执行失败: {text1}')
        self.assertTrue(res2, f'执行失败: {text2}')

        text = f'-----{os.path.basename(__file__)} end-----'
        self.log.info(text)
